//! # 数据源和数据汇单元测试
//! 
//! 测试各种数据源和数据汇的功能

use std::time::Duration;
use std::sync::Arc;
use tokio::time::sleep;
use serde_json::json;
use tempfile::tempdir;

use rustcloud_stream::prelude::*;
use rustcloud_bus::memory::InMemoryMessageBus;
use rustcloud_bus::message::MessageBuilder;

#[tokio::test]
async fn test_memory_stream_source() {
    let mut source = MemoryStreamSource::new("test_memory_source");
    
    // 添加测试消息
    let messages = vec![
        MessageBuilder::new()
            .with_topic("test1")
            .with_payload(json!({"id": 1}))
            .build(),
        MessageBuilder::new()
            .with_topic("test2")
            .with_payload(json!({"id": 2}))
            .build(),
    ];
    
    source.add_messages(messages);
    
    // 测试读取消息
    assert!(!source.is_exhausted());
    
    let msg1 = source.next().await;
    assert!(msg1.is_some());
    let msg1 = msg1.unwrap().unwrap();
    assert_eq!(msg1.payload()["id"], json!(1));
    
    let msg2 = source.next().await;
    assert!(msg2.is_some());
    let msg2 = msg2.unwrap().unwrap();
    assert_eq!(msg2.payload()["id"], json!(2));
    
    // 测试源耗尽
    let msg3 = source.next().await;
    assert!(msg3.is_none());
    assert!(source.is_exhausted());
    
    // 检查统计信息
    let stats = source.get_stats().await;
    assert_eq!(stats.total_received, 2);
    assert_eq!(stats.successful, 2);
    assert_eq!(stats.failed, 0);
}

#[tokio::test]
async fn test_bus_stream_source() {
    let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    let mut source = BusStreamSource::with_defaults(bus.clone(), "test.*").await.unwrap();
    
    // 启动数据源
    source.start().await.unwrap();
    
    // 发布测试消息
    let publisher = bus.create_publisher().unwrap();
    let test_messages = vec![
        MessageBuilder::new()
            .with_topic("test.event1")
            .with_payload(json!({"type": "event", "id": 1}))
            .build(),
        MessageBuilder::new()
            .with_topic("test.event2")
            .with_payload(json!({"type": "event", "id": 2}))
            .build(),
    ];
    
    for msg in test_messages {
        publisher.publish(msg).await.unwrap();
    }
    
    // 等待消息被接收
    sleep(Duration::from_millis(100)).await;
    
    // 从数据源读取消息
    let received1 = source.next().await;
    assert!(received1.is_some());
    let msg1 = received1.unwrap().unwrap();
    assert_eq!(msg1.payload()["id"], json!(1));
    
    let received2 = source.next().await;
    assert!(received2.is_some());
    let msg2 = received2.unwrap().unwrap();
    assert_eq!(msg2.payload()["id"], json!(2));
    
    // 检查统计信息
    let stats = source.get_stats().await;
    assert!(stats.total_received >= 2);
    
    // 停止数据源
    source.stop().await.unwrap();
}

#[tokio::test]
async fn test_timer_stream_source() {
    let mut counter = 0;
    let mut source = TimerStreamSource::new(
        "timer_test",
        Duration::from_millis(50),
        move || {
            counter += 1;
            MessageBuilder::new()
                .with_topic("timer.tick")
                .with_payload(json!({
                    "timestamp": chrono::Utc::now().timestamp(),
                    "tick": counter
                }))
                .build()
        },
    );
    
    // 启动定时器
    source.start().await.unwrap();
    
    // 等待生成消息
    sleep(Duration::from_millis(120)).await;
    
    // 应该至少有两个tick
    let msg1 = source.next().await;
    assert!(msg1.is_some());
    let msg1 = msg1.unwrap().unwrap();
    assert_eq!(msg1.topic(), "timer.tick");
    
    let msg2 = source.next().await;
    assert!(msg2.is_some());
    let msg2 = msg2.unwrap().unwrap();
    assert_eq!(msg2.topic(), "timer.tick");
    
    // 检查统计信息
    let stats = source.get_stats().await;
    assert!(stats.total_received >= 2);
    assert_eq!(stats.successful, stats.total_received);
    
    // 停止定时器
    source.stop().await.unwrap();
}

#[tokio::test]
async fn test_memory_stream_sink() {
    let mut sink = MemoryStreamSink::new("test_memory_sink");
    
    let test_messages = vec![
        MessageBuilder::new()
            .with_topic("output1")
            .with_payload(json!({"result": "success", "id": 1}))
            .build(),
        MessageBuilder::new()
            .with_topic("output2")
            .with_payload(json!({"result": "error", "id": 2}))
            .build(),
    ];
    
    // 发送消息
    for msg in test_messages {
        sink.send(msg).await.unwrap();
    }
    
    // 检查消息
    let messages = sink.get_messages();
    assert_eq!(messages.len(), 2);
    assert_eq!(messages[0].payload()["id"], json!(1));
    assert_eq!(messages[1].payload()["id"], json!(2));
    
    // 检查统计信息
    let stats = sink.get_stats().await;
    assert_eq!(stats.total_sent, 2);
    assert_eq!(stats.successful, 2);
    assert_eq!(stats.failed, 0);
    
    // 测试flush和close
    sink.flush().await.unwrap();
    sink.close().await.unwrap();
}

#[tokio::test]
async fn test_bus_stream_sink() {
    let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    let mut sink = BusStreamSink::with_defaults(bus.clone(), "output.topic").await.unwrap();
    
    // 创建订阅器来验证消息
    let subscriber = bus.subscribe("output.topic").await.unwrap();
    
    // 发送测试消息
    let test_msg = MessageBuilder::new()
        .with_topic("input.topic") // 这个topic会被重写为output.topic
        .with_payload(json!({"data": "test_data", "timestamp": chrono::Utc::now().timestamp()}))
        .build();
    
    sink.send(test_msg).await.unwrap();
    sink.flush().await.unwrap();
    
    // 验证消息是否发送到正确的topic
    let received = subscriber.receive().await.unwrap();
    assert!(received.is_some());
    let msg = received.unwrap();
    assert_eq!(msg.topic(), "output.topic");
    assert_eq!(msg.payload()["data"], json!("test_data"));
    
    // 检查统计信息
    let stats = sink.get_stats().await;
    assert_eq!(stats.total_sent, 1);
    assert_eq!(stats.successful, 1);
    
    sink.close().await.unwrap();
}

#[tokio::test]
async fn test_log_stream_sink() {
    let mut sink = LogStreamSink::new("test_log_sink", "info");
    
    let test_msg = MessageBuilder::new()
        .with_topic("log.test")
        .with_payload(json!({
            "level": "info",
            "message": "This is a test log message",
            "service": "test_service"
        }))
        .build();
    
    // 发送消息到日志
    sink.send(test_msg).await.unwrap();
    sink.flush().await.unwrap();
    
    // 检查统计信息
    let stats = sink.get_stats().await;
    assert_eq!(stats.total_sent, 1);
    assert_eq!(stats.successful, 1);
    assert_eq!(stats.failed, 0);
    
    sink.close().await.unwrap();
}

#[tokio::test]
async fn test_batch_sink_behavior() {
    let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    let config = SinkConfig {
        name: "batch_test".to_string(),
        batch_size: 3,
        buffer_size: 10,
        flush_interval: Duration::from_secs(1),
        timeout: Duration::from_secs(5),
        retry_count: 2,
        retry_interval: Duration::from_millis(100),
    };
    
    let mut sink = BusStreamSink::new("batch_sink", bus.clone(), "batch.output", config).await.unwrap();
    let subscriber = bus.subscribe("batch.output").await.unwrap();
    
    // 发送少于批大小的消息，不应该立即发送
    let msg1 = MessageBuilder::new()
        .with_topic("input")
        .with_payload(json!({"batch": 1}))
        .build();
        
    sink.send(msg1).await.unwrap();
    
    // 检查是否还没有消息发送
    sleep(Duration::from_millis(10)).await;
    let received = subscriber.receive().await.unwrap();
    assert!(received.is_none()); // 应该没有消息
    
    // 发送更多消息直到达到批大小
    for i in 2..=3 {
        let msg = MessageBuilder::new()
            .with_topic("input")
            .with_payload(json!({"batch": i}))
            .build();
        sink.send(msg).await.unwrap();
    }
    
    // 现在应该有消息发送
    sleep(Duration::from_millis(50)).await;
    
    // 验证批处理消息
    let mut received_count = 0;
    for _ in 0..3 {
        if let Some(_msg) = subscriber.receive().await.unwrap() {
            received_count += 1;
        }
    }
    
    assert_eq!(received_count, 3);
    
    // 检查统计信息
    let stats = sink.get_stats().await;
    assert_eq!(stats.total_sent, 3);
    
    sink.close().await.unwrap();
}

#[tokio::test]
async fn test_source_sink_integration() {
    let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    
    // 创建数据源和数据汇
    let mut source = BusStreamSource::with_defaults(bus.clone(), "integration.input").await.unwrap();
    let mut sink = BusStreamSink::with_defaults(bus.clone(), "integration.output").await.unwrap();
    
    // 启动数据源
    source.start().await.unwrap();
    
    // 创建验证订阅器
    let output_subscriber = bus.subscribe("integration.output").await.unwrap();
    
    // 发布输入消息
    let publisher = bus.create_publisher().unwrap();
    let input_msg = MessageBuilder::new()
        .with_topic("integration.input")
        .with_payload(json!({"pipeline": "test", "data": "integration_test"}))
        .build();
    
    publisher.publish(input_msg).await.unwrap();
    
    // 等待消息被接收
    sleep(Duration::from_millis(50)).await;
    
    // 从数据源读取并发送到数据汇
    if let Some(Ok(msg)) = source.next().await {
        sink.send(msg).await.unwrap();
        sink.flush().await.unwrap();
    }
    
    // 验证输出
    let output_msg = output_subscriber.receive().await.unwrap();
    assert!(output_msg.is_some());
    let msg = output_msg.unwrap();
    assert_eq!(msg.topic(), "integration.output");
    assert_eq!(msg.payload()["pipeline"], json!("test"));
    assert_eq!(msg.payload()["data"], json!("integration_test"));
    
    // 停止数据源
    source.stop().await.unwrap();
    sink.close().await.unwrap();
}

#[tokio::test]
async fn test_error_handling_in_sink() {
    // 测试配置错误的情况下的错误处理
    let bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
    
    // 使用无效配置创建sink（超时时间为0）
    let mut config = SinkConfig::default();
    config.timeout = Duration::from_millis(1); // 极短的超时时间
    
    let mut sink = BusStreamSink::new("error_test", bus.clone(), "error.output", config).await.unwrap();
    
    let msg = MessageBuilder::new()
        .with_topic("test")
        .with_payload(json!({"test": "error_handling"}))
        .build();
    
    // 即使有错误，send方法也应该正常返回（因为是异步批处理）
    let result = sink.send(msg).await;
    assert!(result.is_ok());
    
    // flush可能会遇到错误，但在这个简化的测试中应该还是成功的
    let flush_result = sink.flush().await;
    assert!(flush_result.is_ok());
    
    sink.close().await.unwrap();
}